//package com.zzl.kafkatest;
//
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.apache.kafka.clients.consumer.KafkaConsumer;
//import org.apache.kafka.common.serialization.StringDeserializer;
//
//import java.util.Map;
//import java.util.Properties;
//
//public class MyConsumer1 {
//
//    public static void main(String[] args) throws Exception {
//        Properties props = new Properties();
//        props.put("bootstrap.servers", "realtime-1:9092");
//        props.put("group.id", "zzl3");
//        props.put("enable.auto.commit", "false");
//        props.put("auto.commit.interval.ms", "1000");
//        props.put("session.timeout.ms", "30000");
//        props.put("auto.offset.reset", "smallest");
//        props.put("partition.assignment.strategy", "roundrobin");
//
//        props.put("key.deserializer", StringDeserializer.class.getName());
//        props.put("value.deserializer", StringDeserializer.class.getName());
//        KafkaConsumer consumer = new KafkaConsumer<String, String>(props);
//        consumer.subscribe(MyProducer.TOPIC);
//        int messageNo = 1;
//        for (;;){
//            Map<String, ConsumerRecord> poll = consumer.poll(1000);
//            if(null!=poll&&poll.size()>0) {
//                ConsumerRecord record = poll.get(MyProducer.TOPIC);
//                String format = String.format("messageN=%d\nreceive:key = %s\nvalue=%\noffset=%d\npartition=%d\n\n",messageNo, record.key(), record.value(), record.offset(), record.partition());
//                System.out.println(format);
//                messageNo++;
//            }
//        }
//
////        consumer.position()
//    }
//
//}
